package com.ebspp.server.resource;

import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;


import com.ebspp.server.replication.ReplicationManager;
import com.ebspp.server.replication.WatchConnection;
import com.ebspp.server.storage.BDBStore;
import com.ebspp.server.storage.IStore;
import com.ebspp.server.storage.ResourceInitException;
import com.ebspp.server.storage.SimpleMemStore;
import com.ebspp.server.storage.TxnBDBStore;
import com.ebspp.thrift.DataMsg;
import com.ebspp.thrift.Identity;
import com.ebspp.thrift.ResourceType;

/**
 * The class manages the resources provisioned by a client.
 * Which resources are used by the client and how much.
 * @author x23pan
 *
 */
public class ClientResManager
{
	private Map<ResourceType, Integer> resMap = null;
	private Map<ResourceType, IStore> resStoreMap = null;
	private Map<ResourceType, Set<String>> resWatcherMap = null;
	
	private Identity id = null;
	
	IStore memStore = null;
	IStore dbStore = null;
	IStore txnDBStore = null;
	Object mutex = new Object();
	
	public ClientResManager(Identity id)
	{
		this.id = id;
		
		resStoreMap = new HashMap<ResourceType, IStore>();
		resWatcherMap = Collections.synchronizedMap(new HashMap<ResourceType, Set<String>>());
		
		resMap = new HashMap<ResourceType, Integer>();
		for(ResourceType type : ResourceType.values())
			resMap.put(type, 0);
	}
	
	/**
	 * Whether the provision updates is legal
	 * The client should not release more resources than it has
	 * @param updates
	 * @return
	 */
	public boolean canAlloc(Map<ResourceType, Integer> updates)
	{
		for(ResourceType type : updates.keySet())
			if(updates.get(type) + resMap.get(type) < 0)
				return false;
		
		return true;
	}
	
	//perform the provision update
	public void alloc(Map<ResourceType, Integer> updates)
	{
		for(ResourceType type : updates.keySet())
			resMap.put(type, resMap.get(type) + updates.get(type));
	}
	
	//store the new data to the provisioned resource
	public List<DataMsg> write(List<DataMsg> data, ResourceType type)
	{
		if(resMap.get(type) == null)
			return data;
		
		IStore store = null;
		try {
			store = getResource(type);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			System.err.println("Cannot init the resource for write: " + type);
			e.printStackTrace();
			return data;
		}
		
		List<DataMsg> failWrites = store.bulkWrite(data);
		data.removeAll(failWrites);
		
		Set<String> watchers = resWatcherMap.get(type);
		if(watchers != null && watchers.size() > 0)
		{
			for(String watcher : watchers)
				if(!ReplicationManager.notifyWatcher(watcher, type, data))
					System.err.println("Unable to replicate to watcher: " + watcher + ":" + type);
		}
		
		return failWrites;
	}
	
	//read the data back from the server
	public List<DataMsg> read(List<DataMsg> keys, ResourceType type)
	{
		if(resMap.get(type) == null)
			return keys;
		
		IStore store = null;
		try {
			store = getResource(type);
		} 
		catch (Exception e) {
			// TODO Auto-generated catch block
			System.err.println("Cannot init the resource for read: " + type);
			e.printStackTrace();
			return keys;
		}
		
		return store.bulkRead(keys);
	}
	
	public List<DataMsg> garbageCollect(List<DataMsg> keys, ResourceType type)
	{
		if(resMap.get(type) == null)
			return keys;
		
		IStore store = null;
		try {
			store = getResource(type);
		} 
		catch (Exception e) {
			// TODO Auto-generated catch block
			System.err.println("Cannot init the resource for garbage collection: " + type);
			e.printStackTrace();
			return keys;
		}
		
		return store.garbageCollect(keys);
	}
	//a utility function. Given the resource type, return the resources
	private IStore getResource(ResourceType type) throws ResourceInitException
	{
		switch(type)
		{
		case MEMORY:
			if(memStore == null)
			{
				synchronized(mutex)
				{
					if(memStore != null)
						return memStore;
					memStore = new SimpleMemStore();
					resStoreMap.put(type, memStore);
				}
			}
			return resStoreMap.get(type);
		case STORAGE:
			if(dbStore == null)
			{
				synchronized(mutex)
				{
					if(dbStore != null)
						return dbStore;
					dbStore = new BDBStore(id);
					resStoreMap.put(type, dbStore);
				}
			}
			return resStoreMap.get(type);
		case TXN_STORAGE:
			if(txnDBStore == null)
			{
				synchronized(mutex)
				{
					if(txnDBStore != null)
						return txnDBStore;
					txnDBStore = new TxnBDBStore(id);
					resStoreMap.put(type, txnDBStore);
				}
			}
			return resStoreMap.get(type);
		default:
			System.out.println("Unsupported");
			return null;
		}
	}
	
	public boolean watch(ResourceType resType, String watcher)
	{
		WatchConnection waConn = ReplicationManager.createWatchConn(watcher, resType, id);
		if(waConn != null)
		{
			Set<String> watchers = resWatcherMap.get(resType);
			if(watchers == null)
			{
				watchers = new HashSet<String>();
				resWatcherMap.put(resType, watchers);
			}
			watchers.add(watcher);
			return true;
		}
		else
			return false;
	}
	
	public boolean deWatch(ResourceType resType, String watcher)
	{
		boolean deWatch = ReplicationManager.closeWatchConn(watcher, resType);
		Set<String> watchers = resWatcherMap.get(resType);
		if(deWatch && watchers != null)
		{
			watchers.remove(watcher);
			if(watchers.size() == 0)
				resWatcherMap.remove(resType);
		}
		
		return deWatch;
	}
}



/**
 * This thread is used to monitor the client resource usage condition
 * If the usage of the client exceed its provisioned amount, it will try to interfere
 * 
 * @author x23pan
 *
 */
class ResMonitor implements Runnable
{
	ClientResManager cliRes = null;
	
	public ResMonitor(ClientResManager cliRes)
	{
		this.cliRes = cliRes;
	}
	@Override
	public void run() 
	{
		// TODO Auto-generated method stub
		while(true)
		{
			
			try {
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}
	}
}
